--T日统计T-1日03:00:00到 02:59:59的数据，重刷T-3日03:00:00到 02:59:59的数据
--1.发件扫描2.集散发件下一站为中心3.网点发件下一站为中心或者集散（取首条发件）
--set hive.mapred.mode ='proceed'
with
    network_detail as ( --网点和中心的交派关系
        SELECT distri.network_id        --网点编码
             , net.code as network_code --网点code
             , net.name as network_name --网点name
             , distri.can_ex_codes       --可派货集散编码
             , distri.can_ex_names      --可派货集散名称
             , distri.distribution_type
             , net.network_type
        FROM (
                 select network_id               --网点编码
                      , collect_set(code) as      can_ex_codes --可派货集散编码
                      , collect_set(name) as      can_ex_names --可派货集散名称
                      ,distribution_type
                 from jms_dim.dim_lmdm_sys_network_distributi
                      --where distribution_type = 4 --1可派货中心2可交货中心3可交货集散4可派货集散
                 group by network_id,distribution_type
             ) distri
                 inner join
             jms_dim.dim_network_whole_massage net on (net.id = distri.network_id )
    ),
--驿站与网点对应关系
    agent_point as (--门店关联code.用来判断入仓和出仓网点为空,这取这个门店对应的网点code
        select courier_station_code
             , network_code
             , network_name
             , business_id
        from jms_dim.dim_tab_agent_point_base
        where courier_station_code is not null
          and business_id is not null
        group by courier_station_code
               , network_code
               , network_name
               , business_id
    ),
--包号和芯片号的关系
    tab_bag_flight_num as (
        select waybill_no,
               max(flight_num) as flight_num  --芯片号
        from jms_dwd.dwd_tab_barscan_bagging_base_hi
        where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -7) and date_add('{{ execution_date | cst_ds }}',1)
          and substr(waybill_no,1,1)='B'
          and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
        group by waybill_no
    ),
--实际建包扫描(存在一些问题 用时间判断退转件时间和建包扫描时间对比得到退转件之后的建包扫描)
    tab_bagging as (
        select t1.waybill_no,
               t1.scan_site_code,
               t1.scan_site,
               t1.scan_time,                         --建包扫描时间
               t1.scan_user_code,                    --建包扫描人员编码
               t1.scan_user,                         --建包扫描人员名称
               t1.package_code as package_code ,     --建包包号
               t2.flight_num   --芯片号
        from (
                 select waybill_no,
                        scan_site_code,
                        scan_site,
                        scan_time,
                        package_code,
                        scan_user_code,
                        scan_user,
                        row_number() over(partition by waybill_no, scan_site_code order by scan_time) rk
                 from jms_dwd.dwd_tab_barscan_bagging_base_hi
                 where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -7) and date_add('{{ execution_date | cst_ds }}',1)
                   and substr(waybill_no,1,1)!='B'
                   and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
             ) t1 left join tab_bag_flight_num t2 on t1.package_code=t2.waybill_no  --获取包的芯片号 此处只是判断有没有芯片号，不计较是否正确
        where t1.rk = 1
    ),
    base_send_bag as (
        select
            t2.waybill_no,
            t2.scan_site_code,
            t2.scan_site,
            t2.network_type,
            t2.next_network_type,
            t2.next_station_code,
            t2.scan_time,
            t2.package_code,
            tab_bagging.scan_time as bagging_scan_time,                      --建包扫描时间
            tab_bagging.scan_user_code as bagging_user_code,                 --建包扫描人code
            tab_bagging.scan_user as bagging_user,                           --建包扫描人
            tab_bagging.flight_num ,--芯片号
            tab_bagging.package_code as bag_package_code --建包包号
        from
            (
                select * from (
                                  select waybill_no,
                                         scan_site_code,
                                         scan_site,
                                         network_type,
                                         next_network_type,
                                         next_station_code,
                                         scan_time,
                                         package_code,
                                         ROW_NUMBER() over(partition by waybill_no,scan_site_code order by scan_time) rk
                                  from jms_dwd.dwd_tab_barscan_sitesend_base_hi
                                  where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -5) and date_add('{{ execution_date | cst_ds }}',1)
                                    and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
                              ) t where t.rk = 1
                                    --and t.scan_time between concat(date_add('{{ execution_date | cst_ds }}', -2), ' 00:00:00') and concat(date_add('{{ execution_date | cst_ds }}',1), ' 02:59:59')
                                    --扫描网点类型，1.加盟商,2.转运中心,3.集散点,4.普通网点,5.代理区,6.总部,0.其他
                                    and ((t.network_type = 3 and t.next_network_type = 2) --集散发件下一站为中心
                        or (t.network_type = 4 and (t.next_network_type = 3 or t.next_network_type = 2)) --网点发件下一站为中心或者集散
                        or (t.network_type = 3 and t.next_network_type = 3)) --集散发件下一站为集散
                                    and substr(waybill_no,1,1)!='B'
            )t2 left join tab_bagging on t2.waybill_no=tab_bagging.waybill_no  and t2.scan_site_code=tab_bagging.scan_site_code
    ),
    tab_sitesend as (
        select t.waybill_no,
               t.scan_site_code,
               t.scan_site,
               t.network_type,
               t.next_network_type,
               t.scan_time,
               t.next_station_code,
               t.package_code, --包号
               if(b.proxy_package_organ_code is null ,0,1) is_send_poxy, --是否发往代建包机构
               t.bagging_scan_time, --建包扫描时间
               t.bagging_user_code,--建包扫描人code
               t.bagging_user,--建包扫描人
               t.flight_num ,--芯片号
               t.bag_package_code, --建包包号
               if(bd.bagging_scan_time is null,0,1) as is_up_site_baged --上游是否已建包
        from base_send_bag t
                 left join
             (--代建包关系 出港发件可以省略
                 select
                     should_package_organ_code,
                     proxy_package_organ_code,
                     nvl(invalid_time,'9999-12-31 00:00:00') invalid_time,
                     audit_time
                 from  jms_dwd.dwd_tab_proxy_package_base_dt
                 where dt='{{ execution_date | cst_ds }}' and (state != 2 and state !=1)    -- 1 审核中、 2 已驳回、 3 已同意、 4 已失效
             ) b
             on  b.should_package_organ_code =t.scan_site_code and b.proxy_package_organ_code=t.next_station_code  and b.invalid_time>t.scan_time and  t.scan_time>b.audit_time
                 --集散1到集散2到中心 集散1是否有建包
                 left join  base_send_bag bd on bd.network_type in (3,4) and bd.next_network_type = 3  and  t.network_type = 3
            and  bd.next_station_code = t.scan_site_code
            and  bd.waybill_no = t.waybill_no
            --只要集散到集散到中心 的集散到集散
                 left join  base_send_bag bd2 on bd2.network_type = 3 and bd2.next_network_type = 2  and  t.network_type = 3 and t.next_network_type = 3
            and t.next_station_code = bd2.scan_site_code
            and bd2.waybill_no = t.waybill_no
        where  t.scan_time between concat(date_add('{{ execution_date | cst_ds }}', -2), ' 00:00:00') and concat(date_add('{{ execution_date | cst_ds }}',1), ' 02:59:59')
          and ((t.network_type = 3 and t.next_network_type = 2) --集散发件下一站为中心
            or (t.network_type = 4 and (t.next_network_type = 3 or t.next_network_type = 2)) --网点发件下一站为中心或者集散
            or (t.network_type = 3 and t.next_network_type = 3 and bd2.waybill_no is not null))
    ),
--select * from tab_sitesend
--实际扫描的揽收网点
    tab_taking as (
        select t.waybill_no,
               t.scan_site_code,
               t.scan_site,
               t.scan_time,
               net2.can_ex_codes as can_codes_t1  --揽收网点对应的可交货中心
        from (
                 select waybill_no,
                        scan_site_code,
                        input_site as scan_site,
                        scan_time,
                        ROW_NUMBER()  over(partition by waybill_no order by scan_time) rk
                 from jms_dwd.dwd_tab_barscan_taking_base_hi
                 where substr(dt,1,10) between date_add('{{ execution_date | cst_ds }}', -14) and date_add('{{ execution_date | cst_ds }}',1)
                   and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
             ) t
                 left join network_detail net2 on (net2.distribution_type='2' and  net2.network_code=t.scan_site_code) --获取可揽收网点的可交货集散
        where rk = 1
    ),
--实际扫描的签收网点（出库+签收）
    tab_sign as (
        select tt.waybill_no,
               tt.scan_site_code,
               tt.scan_site,
               tt.scan_time,
               tt.post_station_code, --驿站编码
               po.network_code as post_network_code,  --驿站对应网点编码
               po.network_name as post_network_name,   --驿站对于对应网点名称
               net2.can_ex_codes as can_codes_d1
        from (
                 select waybill_no,
                        scan_site_code,
                        scan_site,
                        scan_time,
                        post_station_code,
                        real_oper_source_code,
                        row_number() over(partition by waybill_no order by scan_time) as rk
                 from (
                          select waybill_no,                           --运单
                                 signed_site_code as scan_site_code,   --出库网点
                                 signed_site      as scan_site,        --出库网点name
                                 signed_time      as scan_time,
                                 null             as post_station_code, --驿站编码
                                 null             as real_oper_source_code --实操作来源编码
                          from jms_dwd.dwd_tab_barscan_sign_base_hi
                          where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -2) and date_add('{{ execution_date | cst_ds }}',7)
                            and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
                          union ALL
                          select waybill_no,       --运单
                                 scan_site_code,   --出库网点
                                 scan_site,        --出库网点name
                                 scan_time,
                                 post_station_code, --驿站编码
                                 real_oper_source_code
                          from jms_dwd.dwd_tab_barscan_collect_base_hi
                          where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -2) and date_add('{{ execution_date | cst_ds }}',7)
                            and scan_type = '出库扫描'
                            and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
                      ) t
             ) tt
                 left join network_detail net2 on (net2.distribution_type='1' and  net2.network_code=tt.scan_site_code)
                 left join agent_point po on (tt.real_oper_source_code=po.business_id and tt.post_station_code = po.courier_station_code)
        where tt.rk = 1
    ),
    --实际扫描的出仓网点
    tab_delivery as (
        select t.waybill_no,
               t.scan_site_code,
               t.scan_site,
               t.scan_time
        from (
                 select waybill_no,
                        scan_site_code,
                        scan_site,
                        scan_time,
                        row_number() over(partition by waybill_no order by scan_time) rk
                 from jms_dwd.dwd_tab_barscan_deliver_base_hi
                 where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -2) and date_add('{{ execution_date | cst_ds }}',7)
                   and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
             ) t
        where rk = 1
    ),



    tab_waybill as (
        select waybill_no                           , --运单号
               ordersource_code order_source_code,                                   ---订单来源编码
               ordersource_name order_source_name,                                   --订单来源名称
               goods_type_code,                                     --商品类型
               goods_type_name,
               cast(package_length as decimal(16,2))    package_length , --包裹长
               cast(package_wide as decimal(16,2))      package_wide,  --包裹宽
               cast(package_high as decimal(16,2))      package_high,  --包裹高
               package_inside_charge_weight as package_charge_weight,  --包裹计费重
               customer_code,                                       --客户名称
               customer_name,                                       --客户编码
               dispatch_network_code,                               ---派件网点
               dispatch_network_name,                               --派件网点名称
               package_inside_charge_weight      --内部计费重
        from jms_dwd.dwd_yl_oms_oms_waybill_incre_hi
        where substr(dt, 1, 10) BETWEEN date_add ('{{ execution_date | cst_ds }}', -30) and date_add('{{ execution_date | cst_ds }}',1)
    ),

    tab_order as  --订单表取三段码和实际揽收网点
        (
            select
                t.waybill_id
                 ,t.terminal_dispatch_code
                 ,t.first_code
                 ,f.center_code as end_center_code
                 ,f.center_name as end_center_name
                 ,t.real_pick_network_code
                 ,net1.name as real_pick_network_name
            from
                (
                    select   --订单表取实际实际取件网点和三段码
                        orders.waybill_id
                         ,orders.terminal_dispatch_code
                         ,split(orders.terminal_dispatch_code,',')[0] as first_code
                         ,nvl(real_pick_network_code,pick_network_code) as real_pick_network_code
                    from jms_dwd.dwd_yl_oms_oms_order_dt orders
                    where substr(dt, 1, 10) BETWEEN date_add ('{{ execution_date | cst_ds }}', -30) and date_add('{{ execution_date | cst_ds }}',0)
                      and date(input_time) between date_add ('{{ execution_date | cst_ds }}', -30) and date_add('{{ execution_date | cst_ds }}',0)
union all
select   --订单表取实际实际取件网点和三段码
    orders.waybill_id
     ,orders.terminal_dispatch_code
     ,split(orders.terminal_dispatch_code,',')[0] as first_code
     ,nvl(real_pick_network_code,pick_network_code) as real_pick_network_code
from jms_dwd.dwd_yl_oms_oms_order_hf orders
where dt between concat(date_add('{{ execution_date | cst_ds }}',1),'-00') and concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
  and waybill_rnk=1
  and substr(input_time,1,13) between concat(date_add('{{ execution_date | cst_ds }}',1) ,' 00') and concat(date_add('{{ execution_date | cst_ds }}',1),' 02' )
    )t left join (select code,center_code,center_name from  jms_dim.dim_lmdm_sys_first_code where is_enable=1 and is_delete =1 group by code,center_code,center_name) f on f.code = t.first_code
    left join jms_dim.dim_network_whole_massage net1 on t.real_pick_network_code=net1.code
    ),
    --退转件面单
    tab_reback as (
select *from (
    select t1.waybill_no,                                                    --运单号
    t1.print_flag,                                                    --面单是否已打印
    print_time,
    row_number() over(partition by t1.waybill_no order by print_time) as rk
    from jms_dwd.dwd_tab_reback_transfer_express_base_hi t1
    where
    substr(dt,1,10) >= date_add('{{ execution_date | cst_ds }}', -7)
    and t1.print_flag='1' --面单已经打印
    )t where rk=1
    ),

--将统计需要信息合并
    detail_tmp as (
select
    tab_sitesend.waybill_no,    --运单号
    tab_sitesend.scan_site_code as send_site_code,
    tab_sitesend.scan_site as send_site,
    tab_sitesend.network_type as send_site_type,
    tab_sitesend.next_station_code,
    tab_sitesend.next_network_type ,
    tab_sitesend.scan_time as send_scan_time,
    tab_sitesend.package_code, --包号
    tab_sitesend.is_send_poxy,  --是否发往代建包机构
    tab_waybill.order_source_code,                                   --订单来源编码
    tab_waybill.order_source_name,                                   --订单来源名称
    tab_order.first_code,                                          --运单三段码
    tab_waybill.goods_type_code,                                     --商品类型
    tab_waybill.goods_type_name,
    tab_waybill.package_length,                                      --包裹长
    tab_waybill.package_wide,                                        --包裹宽
    tab_waybill.package_high,                                        --包裹高
    tab_waybill.package_charge_weight,                               --包裹计费重
    tab_waybill.customer_code,                                       --客户名称
    tab_waybill.customer_name,                                       --客户编码
    tab_waybill.dispatch_network_code,                               ---派件网点
    tab_waybill.dispatch_network_name,                               --派件网点名称
--tab_waybill.can_codes_d1 as can_code_fd1,                         --运单中派件网点对应的可派货中心
--tab_waybill.can_codes_d2 as can_code_fd2,                         --运单中揽收网点对应了派货集散对应的可派货中心
    tab_order.real_pick_network_code as pick_network_code,                                   --取件网点
    tab_order.real_pick_network_name as pick_network_name,
    tab_order.end_center_code,                                     --一段码对应的末端中心
    tab_order.end_center_name,                                     --一段码对应的末端中心
--tab_waybill.can_codes_t1 as can_code_ft1,                         --运单中揽收网点对应的可交货中心
--tab_waybill.can_codes_t2 as can_code_ft2,                         --运单中揽收网点对应了交货集散对应的可交货中心
    tab_sign.scan_site_code as sign_scan_code  ,                     --扫描签收网点
    tab_sign.scan_site as sign_scan  ,                               --扫描签收网点
    tab_sign.post_station_code as post_station_code  ,               --驿站编码
    tab_sign.can_codes_d1 as can_code_ad1,                            --扫描签收网点对应的可派货中心
--tab_sign.can_codes_d2 as can_code_ad2  ,                          --扫描签收网点对应的可派货 集散对应的可派货中心
    tab_sign.post_network_code,                                      --驿站绑定的网点编码
    tab_sign.post_network_name,                                      --驿站绑定的网点名称
    tab_taking.scan_site_code as taking_scan_code,                   --扫描揽收网点
    tab_taking.scan_site as taking_scan,                             --扫描揽收网点
    tab_taking.can_codes_t1 as can_code_at1,                          --扫描揽收网点对于应的可交货中心
--tab_taking.can_codes_t2 as can_code_at2,                          --扫描揽收网点对应交货集散对应的可交货中心
    tab_delivery.scan_site_code as delivery_scan_code,               --扫描出仓网点
    tab_delivery.scan_site as delivery_scan,                         --扫描出仓网点
    tab_sitesend.bagging_scan_time,                      --建包扫描时间
    tab_sitesend.bagging_user_code,                 --建包扫描人code
    tab_sitesend.bagging_user,                           --建包扫描人
    tab_reback.waybill_no as reback_waybill_no,                                           --退转件中运单号                            --一段码
    tab_reback.print_time,
    from_unixtime(unix_timestamp(tab_sitesend.scan_time)-3*60*60,'yyyy-MM-dd HH:mm:ss') as send_scan_time3,
    sort_array(array(nvl(tab_waybill.package_length,0),nvl(tab_waybill.package_wide,0),nvl(tab_waybill.package_high,0))) as bian_sort,
    tab_waybill.package_inside_charge_weight, --内部计费重量
    tab_sitesend.flight_num ,--芯片号
    tab_sitesend.bag_package_code, --建包包号
    tab_sitesend.is_up_site_baged --集散A发集散B 集散A是否建包
from tab_sitesend
    left join tab_waybill on tab_sitesend.waybill_no=tab_waybill.waybill_no
    left join tab_sign on tab_sign.waybill_no=tab_sitesend.waybill_no --and tab_taking.scan_time>tab_sitesend.scan_time
    left join tab_taking on tab_taking.waybill_no=tab_sitesend.waybill_no --and tab_taking.scan_time<tab_sitesend.scan_time
    left join tab_delivery on tab_delivery.waybill_no=tab_sitesend.waybill_no
    left join tab_reback on tab_reback.waybill_no=tab_sitesend.waybill_no
    left join tab_order on tab_order.waybill_id=tab_waybill.waybill_no
    )
insert overwrite table jms_tmp.tmp_network_bagging_detail_send_dt partition(dt)
select *, date(send_scan_time3) as dt
from detail_tmp    
distribute by pmod(hash(waybill_no), 90);
----------------------------------------------
insert overwrite table jms_dm.dm_network_bagging_detail_send_dt partition(dt)
select
    waybill_no,
    send_site_code,
    send_site,
    send_site_type,
    next_station_code,
    next_network_type,
    send_scan_time,
    package_code,
    is_send_poxy,
    order_source_code,
    order_source_name,
    first_code,
    goods_type_code,
    goods_type_name,
    package_length,
    package_wide,
    package_high,
    package_charge_weight,
    customer_code,
    customer_name,
    dispatch_network_code,
    dispatch_network_name,
    pick_network_code,
    pick_network_name,
    end_center_code,
    end_center_name,
    sign_scan_code,
    sign_scan,
    post_station_code,
    can_code_ad1,
    post_network_code,
    post_network_name,
    taking_scan_code,
    taking_scan,
    delivery_scan_code,
    delivery_scan,
    bagging_scan_time,
    bagging_user_code,
    bagging_user,
    reback_waybill_no ,
    print_time ,
    send_scan_time3,
    bian_sort,
    is_small,
    is_slender,
    is_own_io,
    count(if(is_small=1 or is_slender=1,waybill_no,null )) over(partition by first_code,substr(send_scan_time3,1,10),send_site_code)  same_bill, --按一段码对小件进行分组单号求和
        is_same_city, --是否是同城件
    package_inside_charge_weight, --内部计费重量
    flight_num, --芯片号
    bag_package_code,
    is_special,
    is_lightweight,
    is_up_site_baged,
    substr(send_scan_time3,1,10) as dt
from
    (
        select
            *,
            if(
--生鲜 液体 易碎品 文件、普通物品类型
                    (nvl(PACKAGE_LENGTH,0) <= 50 and nvl(PACKAGE_WIDE,0) <= 50 and nvl(PACKAGE_HIGH,0) <= 50 --边长不超过30cm
                        and  PACKAGE_CHARGE_WEIGHT <= 3 --所有物品类型<=3
                        and nvl(PACKAGE_LENGTH,0) + nvl(PACKAGE_WIDE,0)+ nvl(PACKAGE_HIGH,0) <= 90
                        )
                ,1,0) as is_small, --是否是小件
            if(nvl(PACKAGE_CHARGE_WEIGHT,0) <=3 and bian_sort[2]<=70 and cast(bian_sort[2]/bian_sort[1] as decimal(16,2))>=5 and PACKAGE_LENGTH+PACKAGE_WIDE+PACKAGE_HIGH<=90
                ,1,0) as is_slender,--是否是细长件
            case when  sign_scan_code is not null     and sign_scan_code=taking_scan_code     then 1
                 when  delivery_scan_code is not null and delivery_scan_code=taking_scan_code then 1
                 when  post_network_code is not null  and post_network_code=taking_scan_code  then 1
                 else 0
                end as is_own_io, --是否是自发自派
            if(can_code_ad1 is not null
                   and end_center_code is not null
                   and can_code_at1 is not null
                   and array_contains(can_code_ad1,end_center_code)
                   and array_contains(can_code_at1,end_center_code) ,1,0)  as is_same_city , --是否同城件
            if(
                            nvl(PACKAGE_CHARGE_WEIGHT,0) <=3
                        and nvl(PACKAGE_LENGTH,0) <= 80 and nvl(PACKAGE_WIDE,0) <= 80 and nvl(PACKAGE_HIGH,0) <= 80
                        and nvl(PACKAGE_LENGTH,0) + nvl(PACKAGE_WIDE,0)+ nvl(PACKAGE_HIGH,0) <= 100
                        and nvl(PACKAGE_LENGTH,0)* nvl(PACKAGE_WIDE,0)* nvl(PACKAGE_HIGH,0)*1000000<=0.03
                ,1,0) as is_special, --是否是特殊件
            if(
                            nvl(PACKAGE_CHARGE_WEIGHT,0) <=0.03
                        and nvl(PACKAGE_LENGTH,0) <= 10 and nvl(PACKAGE_WIDE,0) <= 10 and nvl(PACKAGE_HIGH,0) <= 1
                ,1,0) as is_lightweight --是否轻薄件
        from jms_tmp.tmp_network_bagging_detail_send_dt
--bidefault.dm_network_bagging_detail_send_dt_tmp
        where dt between date_add('{{ execution_date | cst_ds }}',-2) and '{{ execution_date | cst_ds }}'
          and substr(waybill_no,1,1)!='B'
    ) t1
 distribute by pmod(hash(waybill_no), 90);
